package com.beeweeb.services.tps.clustering;

import java.net.Inet4Address;
import java.net.InetAddress;
import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.protocols.FD_ALL;
import org.jgroups.protocols.FD_SOCK;
import org.jgroups.protocols.FRAG2;
import org.jgroups.protocols.MERGE2;
import org.jgroups.protocols.PING;
import org.jgroups.protocols.UDP;
import org.jgroups.protocols.UNICAST2;
import org.jgroups.protocols.VERIFY_SUSPECT;
import org.jgroups.protocols.pbcast.GMS;
import org.jgroups.protocols.pbcast.NAKACK;
import org.jgroups.protocols.pbcast.STABLE;
import org.jgroups.stack.IpAddress;
import org.jgroups.stack.ProtocolStack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author guido d'albore
 */
public class UdpClusterChannel extends JChannel {
    Logger logger = LoggerFactory.getLogger("ThreadPoolScheduler");
    private final static int DEFAULT_MULTICAST_PORT = 45555;
    
    ProtocolStack       stack;
    IpAddress           udpMulticastAddress;
    String              clusterName;
    UdpClusterReceiver  udpClusterReceiver;
    
    public UdpClusterChannel() {
        this("ThreadPoolCluster", "230.30.30.30");
    }

    public boolean isFirstMember() {
//        logger.info("Local address: {}", getAddress());
//        logger.info("Is First Member: {}", ((Address)getView().getMembers().firstElement()).equals(getAddress()));
//        logger.info("Cluster Members:");
//        for(Address a : getView().getMembers()) {
//            logger.info("Member: {}", a.toString());
//        }

        return ((Address)getView().getMembers().firstElement()).equals(getAddress());
    }
    
    public UdpClusterChannel(String clusterName, String udpMulticastAddress) {
        super(false);

        this.clusterName = clusterName;
        setupMulticastAddress(udpMulticastAddress);
        setupProtocolStack();

        udpClusterReceiver = new UdpClusterReceiver();
        super.setReceiver(udpClusterReceiver);
    }

    @Override
    public void disconnect() {
        logger.info("Leaving the cluster '{}'...", clusterName);
        super.disconnect();
        logger.info("Leaving the cluster '{}'...done.", clusterName);
    }

    public void connect() {
        try {
            logger.info("Joining into the cluster '{}'...", clusterName);
            super.setName(UdpClusterNames.getRandomName());
            super.connect(clusterName);
            logger.info("Joining into the cluster '{}'...done.", clusterName);
        } catch(Exception ex) {
            logger.error("Cannot join to cluster. ", ex);
        }


    }
   
    private void setupMulticastAddress(String address) {
        try {
            if(!Inet4Address.getByName(address).isMulticastAddress()) {
                throw new RuntimeException("You can use only valid IPv4 multicast address (from 224.0.0.0 to 239.255.255.255). Check http://en.wikipedia.org/wiki/Multicast_address for details.");
            }

            this.udpMulticastAddress = new IpAddress(address, DEFAULT_MULTICAST_PORT);
        } catch(Exception ex) {
            logger.error("Invalid UDP multicast address.", ex);
        }
    }

    private void setupProtocolStack() {
        // After setup we are ready to prepare the protocol stack
        stack = new ProtocolStack();

        // THIS MUST BE SET BEFORE THE INITIALIZE
        // OTHERWISE YOU'LL GET A NULL POINTER EXCEPTION
        super.setProtocolStack(stack);

        try {
            stack.addProtocol(  (new UDP())
                                .setValue("bind_addr", InetAddress.getByName("0.0.0.0"))
                                .setValue("mcast_addr", udpMulticastAddress)
                                .setValue("mcast_port", DEFAULT_MULTICAST_PORT)
                                .setValue("mcast_recv_buf_size", 100000));  // 100 KB of receive buffer should be enough
        } catch(Exception ex) { }

        stack.addProtocol(  new PING());
        stack.addProtocol(  new MERGE2());
        stack.addProtocol(  new FD_SOCK());

        stack.addProtocol(  (new FD_ALL())
                            .setValue("timeout", 10000)
                            .setValue("interval", 3000)
                            .setValue("max_tries", 5));
        
        stack.addProtocol(  new VERIFY_SUSPECT());
        stack.addProtocol(  new NAKACK());
        stack.addProtocol(  new UNICAST2());
        stack.addProtocol(  new STABLE());
        stack.addProtocol(  new GMS());
        stack.addProtocol(  new FRAG2());

        try {
            stack.init();
        } catch(Exception ex)  {
            logger.error("Something wrong during protocol stack initialization.", ex);
        }
        
    }
}
